package dst.ass3.jms.taskworker.impl;

import java.io.Serializable;
import java.util.logging.Logger;
import javax.annotation.Resource;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.Topic;
import org.apache.openejb.api.LocalClient;
import dst.ass3.dto.ProcessTaskWrapperDTO;
import dst.ass3.jms.Constants;
import dst.ass3.jms.taskworker.ITaskWorker;
import dst.ass3.model.Complexity;
import dst.ass3.model.LifecycleState;

@LocalClient
public class TaskWorker implements ITaskWorker{

	private final static Logger LOGGER = Logger.getLogger(TaskWorker.class.getName());
	
	@Resource(name = Constants.CONNECTION_FACTORY)
    private ConnectionFactory connectionFactory;
	
	@Resource(name = Constants.TASKWORKER_TOPIC)
    private Topic taskworkerTopic;
	
	@Resource(name = Constants.SERVER_QUEUE)
    private Queue serverQ;
	
	private final Complexity complexity;
    private final String taskforce;
    private final String name;
    private ITaskWorkerListener listener;

    private Connection connection;
    private Session session;
    private MessageProducer messageProducer;
    private MessageConsumer messageConsumer;

    public TaskWorker (String name, String taskforce, Complexity complexity) {
        this.name = name;
        this.taskforce = taskforce;
        this.complexity = complexity;
    }
	
	@Override
	public void start() {
		try {
            connection = connectionFactory.createConnection();
            connection.setClientID(name);
            
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            messageProducer = session.createProducer(serverQ);  //send to Server

            //select: pick those that matches taskforce name and specific complexity 
            String selectString = String.format("%s = '%s' AND %s = '%s'",Constants.PROPERTY_TASKFORCE, taskforce, Constants.PROPERTY_COMPLEXITY, complexity);
            
            messageConsumer = session.createDurableSubscriber(taskworkerTopic, name, selectString, false);
            messageConsumer.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message message) {
                    TaskWorker.this.onMessage(message);
                }
            });

            connection.start();
        } catch (JMSException e) {
            e.printStackTrace();
        } 
	}

	protected void onMessage(Message message) {
		if (message == null) {
			LOGGER.info("Message received is null");
            return;
        }
		else{
			ObjectMessage m = (ObjectMessage)message;
	        
	        try {
	            int receivedType = m.getIntProperty(Constants.PROPERTY_MESSAGE_TYPE);
	            
	            LOGGER.info("Message type: "+receivedType+" and object "+ m.getObject());
	            
	            switch (receivedType) {
		            case Constants.MSG_SERVER_TO_TASKWORKE_ASSIGN:
		            	serverAssign(m.getObject());
		                break;
		            default:
		            	LOGGER.info("Type received not known: "+receivedType);
	            }
	        } catch (JMSException e) {
	            e.printStackTrace();
	        }
		}
    }

    private void serverAssign(Serializable object) {
        if (object == null) {
        	LOGGER.info("No object received");
            return;
        }
        else if (listener == null) {
        	LOGGER.info("Listener is null");
            return;
        }
        else{
	        ProcessTaskWrapperDTO dto = (ProcessTaskWrapperDTO)object;
	
	        listener.waitTillProcessed(dto, name, dto.getComplexity(), dto.getRatedBy());
	
	        try {
	            dto.setState(LifecycleState.PROCESSED);
	
	            ObjectMessage msg = session.createObjectMessage(dto);
	            msg.setIntProperty(Constants.PROPERTY_MESSAGE_TYPE, Constants.MSG_TASKWORKER_TO_SERVER_PROCESSED);
	            
	            messageProducer.send(msg);		//send to Server
	        } catch (JMSException e) {
	            e.printStackTrace();
	        }
        }
    }
	
	@Override
	public void stop() {
		try {
			if (messageConsumer != null)
				messageConsumer.close();
		} catch (JMSException e) {
			e.printStackTrace();
		}
		try {
			if (messageProducer != null)
				messageProducer.close();
		} catch (JMSException e) {
			e.printStackTrace();
		}
		try {
			if (session != null)
				session.close();
		} catch (JMSException e) {
			e.printStackTrace();
		}
		try {
			if (connection != null)
				connection.close();
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}

	@Override
	public void setTaskWorkerListener(ITaskWorkerListener listener) {
		this.listener = listener;
	}

}
